/*
 * Copyright (c) 2025 Li Auto Inc. and its affiliates
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "UDSStreamChannelResource.h"
#include "UDSStreamTransportInterface.h"

#include <chrono>
#include <thread>

#include <deps/common/IPLocator.h>

namespace vbs {
namespace transport {
using IPLocator = vbsutil::xmlparser::IPLocator;

UDSStreamChannelResource::UDSStreamChannelResource(UDSStreamTransportInterface* parent, asio::io_service& service,
                                                   const Locator& locator, const Locator& remote_locator,
                                                   uint32_t maxMsgSize)
    : ChannelResource(maxMsgSize),
      service_(service),
      parent_(parent),
      locator_(locator),
      remote_locator_(remote_locator),
      connection_status_(UDSConnectionStatus::eDisconnected),
      connection_type_(UDSConnectionType::UDS_CONNECT_TYPE) {}

UDSStreamChannelResource::UDSStreamChannelResource(UDSStreamTransportInterface* parent, asio::io_service& service,
                                                   const Locator& locator, uint32_t maxMsgSize)
    : ChannelResource(maxMsgSize),
      service_(service),
      parent_(parent),
      locator_(locator),
      connection_status_(UDSConnectionStatus::eDisconnected),
      connection_type_(UDSConnectionType::UDS_CONNECT_TYPE) {}

UDSStreamChannelResource::UDSStreamChannelResource(UDSStreamTransportInterface* parent, asio::io_service& service,
                                                   std::shared_ptr<asio::local::stream_protocol::socket> socket,
                                                   const Locator& locator, const Locator& remote_locator,
                                                   uint32_t maxMsgSize, bool is_server)
    : ChannelResource(maxMsgSize),
      service_(service),
      socket_(std::move(socket)),
      parent_(parent),
      locator_(locator),
      remote_locator_(remote_locator),
      connection_status_(UDSConnectionStatus::eDisconnected),
      connection_type_(UDSConnectionType::UDS_CONNECT_TYPE),
      is_server_(is_server) {}

UDSStreamChannelResource::~UDSStreamChannelResource() {}

void UDSStreamChannelResource::disable() {
    ChannelResource::disable();
    disconnect();
}

bool UDSStreamChannelResource::connect(const std::shared_ptr<UDSStreamChannelResource>& myself) {

    assert(UDSConnectionType::UDS_CONNECT_TYPE == connection_type_);
    UDSConnectionStatus expected = UDSConnectionStatus::eDisconnected;

    if (connection_status_.compare_exchange_strong(expected, UDSConnectionStatus::eConnecting)) {
        try {
            // 创建 socket
            socket_ = std::make_shared<asio::local::stream_protocol::socket>(service_);
            std::weak_ptr<UDSStreamChannelResource> channel_weak_ptr = myself;
            std::string path_s = IPLocator::getPathFromLocator(remote_locator_);
            std::string path_c = IPLocator::getPathFromLocator(locator_);
            socket_->open(asio::local::stream_protocol::socket::protocol_type());
            asio::local::stream_protocol::endpoint endpoint_s(path_s);
            asio::local::stream_protocol::endpoint endpoint_c(path_c);
            if (std::remove(path_c.c_str()) != 0 && errno != ENOENT) {
                logError(UDSSTREAM,
                         "Failed to remove socket when connect!!! file: " << path_c << ", error: " << strerror(errno));
                return false;
            }
            socket_->bind(endpoint_c);

            // 同步连接
            socket_->connect(endpoint_s);

            //重置local_locator
            if (!IPLocator::setPathToLocator((socket_->local_endpoint()).path(), locator_)) {
                logError(UDSSTREAM, "reset local locator failed");
            }

            // 通知连接成功
            if (!channel_weak_ptr.expired()) {
                parent_->SocketConnected(channel_weak_ptr, std::error_code());
            }
        } catch (const asio::system_error& error) {
            // 捕获异常并记录日志
            logError(UDSSTREAM, "Opening socket failed: " << error.what());

            // 恢复状态为断开连接
            connection_status_.store(UDSConnectionStatus::eDisconnected);
            return false;
        }
    }
    return true;
}

void UDSStreamChannelResource::disconnect() {
    logDebug(UDSSTREAM, "UDSStreamChannelResource disconnect status:" << connection_status() << "alive:" << alive());
    if ((eConnecting < change_status(UDSConnectionStatus::eDisconnected)) && alive()) {
        logDebug(UDSSTREAM, "UDSStreamChannelResource disconnect");
        auto socket = socket_;
        std::error_code ec;
        (void)socket->shutdown(asio::local::stream_protocol::socket::shutdown_both, ec);
        std::string path = socket->local_endpoint().path();
        service_.post([&, is_server = is_server_, uds_path = path, socket_m = std::move(socket)]() {
            try {
                socket_m->cancel();
                socket_m->close();
                if (!is_server) {
                    if (std::remove(uds_path.c_str()) != 0 && errno != ENOENT) {
                        logWarning(UDSSTREAM, "Failed to remove socket when disconnect!!!");
                    }
                    logDebug(UDSSTREAM, "is_server:" << is_server << " remove path:" << uds_path);
                }
            } catch (std::exception&) {}
        });
    } else {
        logDebug(UDSSTREAM, "not need disconnect status:" << connection_status() << " alive:" << alive());
    }
}

uint32_t UDSStreamChannelResource::read(ertps::rtps::octet* buffer, std::size_t size, asio::error_code& ec) {
    std::unique_lock<std::mutex> read_lock(read_mutex_);
    if (eConnecting < connection_status_) {
        return static_cast<uint32_t>(
            asio::read(*socket_, asio::buffer(buffer, size), asio::transfer_exactly(size), ec));
    }
    return 0;
}

size_t UDSStreamChannelResource::send(const ertps::rtps::octet* header, size_t header_size,
                                      const ertps::rtps::octet* data, size_t size, asio::error_code& ec) {
    size_t bytes_sent = 0;
    if (eConnecting < connection_status_) {
        std::lock_guard<std::mutex> send_guard(send_mutex_);
        if (header_size > 0) {
            std::array<asio::const_buffer, 2> buffers;
            buffers[0] = asio::buffer(header, header_size);
            buffers[1] = asio::buffer(data, size);
            bytes_sent = asio::write(*socket_.get(), buffers, ec);
        } else {
            bytes_sent = asio::write(*socket_.get(), asio::buffer(data, size), ec);
        }
    }
    return bytes_sent;
}

void UDSStreamChannelResource::async_send(const ertps::rtps::octet* header, size_t header_size,
                                          const ertps::rtps::octet* data, size_t size, AsyncSendHandler handler,
                                          bool is_default_header) {
    if (eConnecting < connection_status_) {
        std::lock_guard<std::mutex> send_guard(send_mutex_);
        auto buffers_ptr = std::make_shared<std::vector<asio::const_buffer>>();

        if (header_size > 0) {
            buffers_ptr->emplace_back(asio::buffer(header, header_size));
        }

        buffers_ptr->emplace_back(asio::buffer(data, size));

        asio::async_write(*socket_, *buffers_ptr,
                          [buffers_ptr, handler_m = std::move(handler), is_default_header, header_size](
                              const asio::error_code& ec, size_t bytes_transferred) {
                              logInfo(UDSSTREAM, "async_write bytes_transferred: " << bytes_transferred);
                              if (handler_m) {
                                  if (is_default_header) {
                                      bytes_transferred -= header_size;
                                  }
                                  handler_m(ec, bytes_transferred);
                              }
                          });
    } else {
        asio::error_code ec = asio::error::not_connected;
        logError(UDSSTREAM, "async_write bytes_transferred: " << 0);
        if (handler) {
            handler(ec, 0);
        }
    }
}

asio::local::stream_protocol::endpoint UDSStreamChannelResource::remote_endpoint() const {
    return socket_->remote_endpoint();
}

asio::local::stream_protocol::endpoint UDSStreamChannelResource::local_endpoint() const {
    std::error_code ec;
    return socket_->local_endpoint(ec);
}

void UDSStreamChannelResource::set_options(const UDSStreamTransportDescriptor* options) {
    socket_->set_option(asio::socket_base::receive_buffer_size(options->receiveBufferSize));
    socket_->set_option(asio::socket_base::send_buffer_size(options->sendBufferSize));
}

void UDSStreamChannelResource::cancel() {
    try {
        socket_->cancel();
    } catch (std::exception&) {}
}

void UDSStreamChannelResource::close() {
    if (socket_) {
        socket_->close();
    }
}

void UDSStreamChannelResource::shutdown(asio::socket_base::shutdown_type what) {
    socket_->shutdown(what);
}

}  // namespace transport
}  // namespace vbs
